package com.atguigu.gmall.apps

import java.time.{LocalDate, LocalDateTime}
import java.time.format.DateTimeFormatter
import java.util

import com.alibaba.fastjson.JSON
import com.atguigu.gmall.Beans.{OrderDetail, OrderInfo, SaleDetail, UserInfo}
import com.atguigu.gmall.utils.Constant.KafkaConstant
import com.atguigu.gmall.utils.{MyESUtil, MykafkaUtil, RedisUtil}
import com.google.gson.Gson
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer

/**
  * @ClassName: SaleApp
  * @Description:
  * @Author: kele
  * @Date: 2021/3/26 14:38
  **/
object SaleApp extends baseApp {

  override val AppName: String = "sale"
  override val time: Int = 10


  def main(args: Array[String]): Unit = {

    runApp{

      //1、从kafka中消费数据
      val od: InputDStream[ConsumerRecord[String, String]] = MykafkaUtil.getKafkaStream(KafkaConstant.KAFKA_TOPIC_ORDER_DETAIL, streamingContext)

      val oi: InputDStream[ConsumerRecord[String, String]] = MykafkaUtil.getKafkaStream(KafkaConstant.KAFKA_TOPIC_ORDER_INFO, streamingContext)

      //2、将获得的数据转化（k,v）
      //获取订单详情
      val od1= od.map(record => {

        val orderDetail: OrderDetail = JSON.parseObject(record.value(), classOf[OrderDetail])

        (orderDetail.order_id,orderDetail)
      })
      //获取订单数据
      val oi1 = oi.map(record=>{

        val orderinfo: OrderInfo = JSON.parseObject(record.value(),classOf[OrderInfo])


        val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
        val formatter2 = DateTimeFormatter.ofPattern("yyyy-MM-dd")

        //将string的日期格式转化为LocalDateTime类型
        val time: LocalDateTime = LocalDateTime.parse(orderinfo.create_time,formatter)

        //获取创建订单的日期
        orderinfo.create_date = time.format(formatter2)

        //获取创建订单的时间
        orderinfo.create_hour = LocalDateTime.parse(orderinfo.create_time,formatter).getHour.toString

        //对手机号前三后四脱敏
        orderinfo.consignee_tel = orderinfo.consignee_tel.replaceAll("(\\d{3})\\d{4}(\\d{4})", "$1****$2");
        //将转化的对象返回
        (orderinfo.id,orderinfo)
      })

      /**
        * 3、双流join
        *     1、先对 订单详情 和 订单表 进行fulljoin
        *     2、处理逻辑
        *
        *     Order_info不为空
        *         1、order_info全部写到redis
        *         2、order_detail不为空
        *         3、order_detail为空
        *             redis中有：合并
        *             redis中没：order_detail数据还没到，等待，不处理
        *
        *    Order_detail不为空（order_info肯定为空）
        *        1、在redis中找order_info
        *        redis中有：合并
        *        redis中没：detail写到redis
        *
        *
        */

      //①、订单表和订单详情表fulljoin
      val f1: DStream[(String, (Option[OrderInfo], Option[OrderDetail]))] = oi1.fullOuterJoin(od1)

      val f2: DStream[SaleDetail] = f1.mapPartitions(iter => {

        //获取redis的连接
        val jedisClient: Jedis = RedisUtil.getJedisClient

        //将封装的对象存在list中
        val saleDetails = new ListBuffer[SaleDetail]()

        //转为json的语句
        var gson = new Gson()

        iter.foreach {

          case (id, (info, detail)) => {

            /**
              * 处理info存在的情况
              * --1、detail存在
              *
              * 将数据写入redis中，防止下次orderdetail需要（orderinfo一对多orderdetail）
              *
              * 判断detail在redis中是否在，
              *
              *
              */
            if (info != None) {

              //info是Option的类型。通过get获取数据
              val orderinfo: OrderInfo = info.get

              if (detail != None) {

                val orderdetail: OrderDetail = detail.get

                //合并数据，将数据封装为saledetail的格式
                val saleDetail = new SaleDetail(orderinfo, orderdetail)
                saleDetails.append(saleDetail)
              }

              /**
                * info不适合使用set存，存在set中不适合设置过期时间，导致数据会越来越大
                * 按照string的类型存，
                * key : order_info + id
                * value : string
                *
                */

              val key: String = "orede_info" + id
              val value: String = gson.toJson(orderinfo)

              //将数据存入jedis中，并设置过期时间
              jedisClient.setex(key, 500, value)

              //在redis中日否有同一个id的detail信息，如果有，就添加
              if (jedisClient.exists("order_detail" + id)) {

                val orderdetail: util.Set[String] = jedisClient.smembers("order_detail" + orderinfo.id)

                import collection.JavaConverters._

                //订单详情一件商品对应一条记录，遍历每个商品组合对应的订单
                //foreach：只要集合是空，不通过foreach，可以省略之前的判断
                orderdetail.asScala.foreach(od => {

                  val detail: OrderDetail = JSON.parseObject(od, classOf[OrderDetail])

                  saleDetails.append(new SaleDetail(orderinfo, detail))
                })

              }

            }

            /**
              * 处理：order_info 不存在，detail存在
              *
              * Order_detail不为空（order_info肯定为空）
              * 1、在redis中找order_info
              * redis中有：合并
              * redis中没：detail写到redis
              */
            else {

              val orderdetail: OrderDetail = detail.get

              if (jedisClient.exists("order_info" + id)) {

                val str: String = jedisClient.get("order_info" + id)

                val orderInfo: OrderInfo = JSON.parseObject(str, classOf[OrderInfo])

                saleDetails.append(new SaleDetail(orderInfo, orderdetail))

              } else {

                val detailjson: String = gson.toJson(orderdetail)

                jedisClient.sadd("orderdetail" + id, detailjson)

                jedisClient.expire("orderdetail" + id, 60 * 60 * 24)

              }

            }

          }
        }

        jedisClient.close()

        saleDetails.iterator
      })



      //用户全量导入在redis中

      val f3: DStream[SaleDetail] = f2.mapPartitions(iter => {

        val jedisClient: Jedis = RedisUtil.getJedisClient

         iter.map(saledetail => {

          val info: String = jedisClient.get("userinfo" + saledetail.user_id)

          val userinfo: UserInfo = JSON.parseObject(info, classOf[UserInfo])

          saledetail.mergeUserInfo(userinfo)

          saledetail

        })



      })


      //将数据写入es中
      f3.foreachRDD(rdd=>{

        rdd.foreachPartition(data=>{

          val date: String = LocalDate.now().toString

          val list: List[(String, SaleDetail)] = data.map(data_info=>{(data_info.order_detail_id,data_info)}).toList


          MyESUtil.insertBulk("gmall2021_sale_detail"+date,list)

        })


      })





    }




  }

}
